package org.apache.flink.state

import org.apache.flink.api.common.functions.{AggregateFunction, RichFlatMapFunction}
import org.apache.flink.api.common.state.{AggregatingState, AggregatingStateDescriptor, MapState, MapStateDescriptor, ReducingState}
import org.apache.flink.api.scala._
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.util.Collector

object KeyedStateApp4 {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.fromElements(
      (1, "buy", "iphone"),
      (1, "cart", "mi"),
      (1, "buy", "spark"),
      (1, "fav", "oppo"),
      (2, "buy", "huawei"),
      (2, "buy", "onemore"),
      (2, "fav", "iphone")
    ).keyBy(_._1)
        .flatMap(new RichFlatMapFunction[(Int,String,String),(Int, String,Int)] {
          private var behaviorCntState: MapState[String,Int] = _

          override def open(parameters: Configuration): Unit = {
            val mapState = new MapStateDescriptor[String,Int]("map",classOf[String],classOf[Int])
            behaviorCntState = getRuntimeContext.getMapState(mapState)
          }


          override def flatMap(value: (Int, String, String), out: Collector[(Int, String, Int)]): Unit = {
            var behaviorCnt = 1

            if(behaviorCntState.contains(value._2)) {
              behaviorCnt = behaviorCntState.get(value._2) + 1
            }

            behaviorCntState.put(value._2, behaviorCnt)
            out.collect((value._1, value._2, behaviorCnt))
          }
        }).print()

    env.execute(getClass.getCanonicalName)
  }
}

